//! # 消息总线配置
//!
//! 定义消息总线系统的配置结构和加载机制

use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::time::Duration;

use crate::error::{MessageBusError, MessageBusResult};

/// 消息总线类型
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum MessageBusType {
    /// 内存消息总线
    Memory,
    /// Redis 消息总线
    Redis,
    /// Kafka 消息总线
    Kafka,
    /// NATS 消息总线
    Nats,
}

impl Default for MessageBusType {
    fn default() -> Self {
        MessageBusType::Memory
    }
}

/// Redis 配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RedisConfig {
    /// Redis 连接 URL
    pub url: String,
    /// 连接池大小
    pub pool_size: u32,
    /// 连接超时时间（毫秒）
    pub connection_timeout_ms: u64,
    /// 命令超时时间（毫秒）
    pub command_timeout_ms: u64,
    /// 数据库索引
    pub database: u8,
    /// 密码
    pub password: Option<String>,
    /// 流的最大长度
    pub stream_max_length: u64,
    /// 消费者组前缀
    pub consumer_group_prefix: String,
}

impl Default for RedisConfig {
    fn default() -> Self {
        Self {
            url: "redis://127.0.0.1:6379".to_string(),
            pool_size: 10,
            connection_timeout_ms: 5000,
            command_timeout_ms: 3000,
            database: 0,
            password: None,
            stream_max_length: 10000,
            consumer_group_prefix: "rustcloud-bus".to_string(),
        }
    }
}

/// Kafka 配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KafkaConfig {
    /// Broker 地址列表
    pub brokers: Vec<String>,
    /// 客户端 ID
    pub client_id: String,
    /// 安全协议
    pub security_protocol: String,
    /// SASL 机制
    pub sasl_mechanism: Option<String>,
    /// SASL 用户名
    pub sasl_username: Option<String>,
    /// SASL 密码
    pub sasl_password: Option<String>,
    /// 生产者配置
    pub producer: KafkaProducerConfig,
    /// 消费者配置
    pub consumer: KafkaConsumerConfig,
}

impl Default for KafkaConfig {
    fn default() -> Self {
        Self {
            brokers: vec!["localhost:9092".to_string()],
            client_id: "rustcloud-bus".to_string(),
            security_protocol: "PLAINTEXT".to_string(),
            sasl_mechanism: None,
            sasl_username: None,
            sasl_password: None,
            producer: KafkaProducerConfig::default(),
            consumer: KafkaConsumerConfig::default(),
        }
    }
}

/// Kafka 生产者配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KafkaProducerConfig {
    /// 确认级别
    pub acks: String,
    /// 重试次数
    pub retries: u32,
    /// 批处理大小
    pub batch_size: u32,
    /// 延迟时间（毫秒）
    pub linger_ms: u64,
    /// 缓冲内存大小
    pub buffer_memory: u64,
    /// 压缩类型
    pub compression_type: String,
}

impl Default for KafkaProducerConfig {
    fn default() -> Self {
        Self {
            acks: "all".to_string(),
            retries: 3,
            batch_size: 16384,
            linger_ms: 0,
            buffer_memory: 33554432, // 32MB
            compression_type: "none".to_string(),
        }
    }
}

/// Kafka 消费者配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KafkaConsumerConfig {
    /// 消费者组 ID
    pub group_id: String,
    /// 自动偏移重置策略
    pub auto_offset_reset: String,
    /// 是否启用自动提交
    pub enable_auto_commit: bool,
    /// 自动提交间隔（毫秒）
    pub auto_commit_interval_ms: u64,
    /// 会话超时时间（毫秒）
    pub session_timeout_ms: u64,
    /// 心跳间隔（毫秒）
    pub heartbeat_interval_ms: u64,
    /// 最大轮询记录数
    pub max_poll_records: u32,
}

impl Default for KafkaConsumerConfig {
    fn default() -> Self {
        Self {
            group_id: "rustcloud-bus-group".to_string(),
            auto_offset_reset: "earliest".to_string(),
            enable_auto_commit: false,
            auto_commit_interval_ms: 5000,
            session_timeout_ms: 30000,
            heartbeat_interval_ms: 3000,
            max_poll_records: 500,
        }
    }
}

/// NATS 配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NatsConfig {
    /// NATS 服务器 URL
    pub url: String,
    /// 连接名称
    pub name: String,
    /// 最大重连次数
    pub max_reconnects: u32,
    /// 重连延迟（毫秒）
    pub reconnect_delay_ms: u64,
    /// 连接超时时间（毫秒）
    pub connection_timeout_ms: u64,
    /// Ping 间隔（毫秒）
    pub ping_interval_ms: u64,
    /// 用户名
    pub username: Option<String>,
    /// 密码
    pub password: Option<String>,
    /// Token
    pub token: Option<String>,
    /// JetStream 配置
    pub jetstream: NatsJetStreamConfig,
}

impl Default for NatsConfig {
    fn default() -> Self {
        Self {
            url: "nats://127.0.0.1:4222".to_string(),
            name: "rustcloud-bus".to_string(),
            max_reconnects: 10,
            reconnect_delay_ms: 2000,
            connection_timeout_ms: 5000,
            ping_interval_ms: 60000,
            username: None,
            password: None,
            token: None,
            jetstream: NatsJetStreamConfig::default(),
        }
    }
}

/// NATS JetStream 配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NatsJetStreamConfig {
    /// 是否启用 JetStream
    pub enabled: bool,
    /// 流名称前缀
    pub stream_prefix: String,
    /// 消费者名称前缀
    pub consumer_prefix: String,
    /// 最大消息数
    pub max_messages: u64,
    /// 最大字节数
    pub max_bytes: u64,
    /// 消息保留策略
    pub retention_policy: String,
    /// 存储类型
    pub storage_type: String,
}

impl Default for NatsJetStreamConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            stream_prefix: "RUSTCLOUD_BUS".to_string(),
            consumer_prefix: "rustcloud-bus".to_string(),
            max_messages: 100000,
            max_bytes: 1073741824, // 1GB
            retention_policy: "limits".to_string(),
            storage_type: "file".to_string(),
        }
    }
}

/// 消息总线配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageBusConfig {
    /// 消息总线类型
    pub bus_type: MessageBusType,
    /// 最大连接数
    pub max_connections: usize,
    /// 连接超时时间（毫秒）
    pub connection_timeout_ms: u64,
    /// 重试次数
    pub retry_attempts: u32,
    /// 重试延迟（毫秒）
    pub retry_delay_ms: u64,
    /// 是否启用指标收集
    pub enable_metrics: bool,
    /// 指标端口
    pub metrics_port: u16,
    /// 是否启用追踪
    pub enable_tracing: bool,
    /// 缓冲区大小
    pub buffer_size: usize,
    /// 消息最大大小（字节）
    pub max_message_size: usize,
    /// 消息 TTL（毫秒）
    pub message_ttl_ms: u64,
    /// 自定义属性
    pub custom_properties: HashMap<String, String>,
    
    /// Redis 特定配置
    pub redis: Option<RedisConfig>,
    /// Kafka 特定配置
    pub kafka: Option<KafkaConfig>,
    /// NATS 特定配置
    pub nats: Option<NatsConfig>,
}

impl Default for MessageBusConfig {
    fn default() -> Self {
        Self {
            bus_type: MessageBusType::default(),
            max_connections: 100,
            connection_timeout_ms: 5000,
            retry_attempts: 3,
            retry_delay_ms: 1000,
            enable_metrics: true,
            metrics_port: 9090,
            enable_tracing: true,
            buffer_size: 1000,
            max_message_size: 1048576, // 1MB
            message_ttl_ms: 3600000, // 1小时
            custom_properties: HashMap::new(),
            redis: None,
            kafka: None,
            nats: None,
        }
    }
}

impl MessageBusConfig {
    /// 从环境变量加载配置
    pub fn from_env() -> MessageBusResult<Self> {
        let mut config = Self::default();
        
        // 消息总线类型
        if let Ok(bus_type) = std::env::var("RUSTCLOUD_BUS_TYPE") {
            config.bus_type = match bus_type.to_lowercase().as_str() {
                "memory" => MessageBusType::Memory,
                "redis" => MessageBusType::Redis,
                "kafka" => MessageBusType::Kafka,
                "nats" => MessageBusType::Nats,
                _ => return Err(MessageBusError::config_error(
                    "RUSTCLOUD_BUS_TYPE", 
                    bus_type, 
                    "不支持的消息总线类型"
                )),
            };
        }
        
        // 最大连接数
        if let Ok(max_connections) = std::env::var("RUSTCLOUD_BUS_MAX_CONNECTIONS") {
            config.max_connections = max_connections.parse::<usize>()
                .map_err(|e: std::num::ParseIntError| MessageBusError::config_error(
                    "RUSTCLOUD_BUS_MAX_CONNECTIONS", 
                    max_connections, 
                    e.to_string()
                ))?;
        }
        
        // 重试次数
        if let Ok(retry_attempts) = std::env::var("RUSTCLOUD_BUS_RETRY_ATTEMPTS") {
            config.retry_attempts = retry_attempts.parse::<u32>()
                .map_err(|e: std::num::ParseIntError| MessageBusError::config_error(
                    "RUSTCLOUD_BUS_RETRY_ATTEMPTS", 
                    retry_attempts, 
                    e.to_string()
                ))?;
        }
        
        // 是否启用指标
        if let Ok(enable_metrics) = std::env::var("RUSTCLOUD_BUS_ENABLE_METRICS") {
            config.enable_metrics = enable_metrics.parse::<bool>()
                .map_err(|e: std::str::ParseBoolError| MessageBusError::config_error(
                    "RUSTCLOUD_BUS_ENABLE_METRICS", 
                    enable_metrics, 
                    e.to_string()
                ))?;
        }
        
        // Redis 配置
        if config.bus_type == MessageBusType::Redis {
            let mut redis_config = RedisConfig::default();
            
            if let Ok(url) = std::env::var("RUSTCLOUD_BUS_REDIS_URL") {
                redis_config.url = url;
            }
            
            if let Ok(pool_size) = std::env::var("RUSTCLOUD_BUS_REDIS_POOL_SIZE") {
                redis_config.pool_size = pool_size.parse::<u32>()
                    .map_err(|e: std::num::ParseIntError| MessageBusError::config_error(
                        "RUSTCLOUD_BUS_REDIS_POOL_SIZE", 
                        pool_size, 
                        e.to_string()
                    ))?;
            }
            
            config.redis = Some(redis_config);
        }
        
        // Kafka 配置
        if config.bus_type == MessageBusType::Kafka {
            let mut kafka_config = KafkaConfig::default();
            
            if let Ok(brokers) = std::env::var("RUSTCLOUD_BUS_KAFKA_BROKERS") {
                kafka_config.brokers = brokers.split(',').map(|s| s.trim().to_string()).collect();
            }
            
            if let Ok(client_id) = std::env::var("RUSTCLOUD_BUS_KAFKA_CLIENT_ID") {
                kafka_config.client_id = client_id;
            }
            
            config.kafka = Some(kafka_config);
        }
        
        // NATS 配置
        if config.bus_type == MessageBusType::Nats {
            let mut nats_config = NatsConfig::default();
            
            if let Ok(url) = std::env::var("RUSTCLOUD_BUS_NATS_URL") {
                nats_config.url = url;
            }
            
            if let Ok(name) = std::env::var("RUSTCLOUD_BUS_NATS_NAME") {
                nats_config.name = name;
            }
            
            config.nats = Some(nats_config);
        }
        
        Ok(config)
    }
    
    /// 从文件加载配置
    pub fn from_file(path: &str) -> MessageBusResult<Self> {
        let content = std::fs::read_to_string(path)
            .map_err(|e| MessageBusError::config_error("file", path, e.to_string()))?;
        
        if path.ends_with(".yaml") || path.ends_with(".yml") {
            serde_yaml::from_str(&content)
                .map_err(|e: serde_yaml::Error| MessageBusError::config_error("yaml", path, e.to_string()))
        } else if path.ends_with(".toml") {
            toml::from_str(&content)
                .map_err(|e: toml::de::Error| MessageBusError::config_error("toml", path, e.to_string()))
        } else if path.ends_with(".json") {
            serde_json::from_str(&content)
                .map_err(|e: serde_json::Error| MessageBusError::config_error("json", path, e.to_string()))
        } else {
            Err(MessageBusError::config_error(
                "format", 
                path, 
                "不支持的配置文件格式，支持 yaml/yml/toml/json"
            ))
        }
    }
    
    /// 验证配置
    pub fn validate(&self) -> MessageBusResult<()> {
        // 验证基本配置
        if self.max_connections == 0 {
            return Err(MessageBusError::config_error(
                "max_connections", 
                "0", 
                "最大连接数必须大于 0"
            ));
        }
        
        if self.buffer_size == 0 {
            return Err(MessageBusError::config_error(
                "buffer_size", 
                "0", 
                "缓冲区大小必须大于 0"
            ));
        }
        
        if self.max_message_size == 0 {
            return Err(MessageBusError::config_error(
                "max_message_size", 
                "0", 
                "最大消息大小必须大于 0"
            ));
        }
        
        // 验证特定配置
        match self.bus_type {
            MessageBusType::Redis => {
                if let Some(ref redis_config) = self.redis {
                    if redis_config.url.is_empty() {
                        return Err(MessageBusError::config_error(
                            "redis.url", 
                            "", 
                            "Redis URL 不能为空"
                        ));
                    }
                    if redis_config.pool_size == 0 {
                        return Err(MessageBusError::config_error(
                            "redis.pool_size", 
                            "0", 
                            "Redis 连接池大小必须大于 0"
                        ));
                    }
                } else {
                    return Err(MessageBusError::config_error(
                        "redis", 
                        "null", 
                        "Redis 配置缺失"
                    ));
                }
            }
            MessageBusType::Kafka => {
                if let Some(ref kafka_config) = self.kafka {
                    if kafka_config.brokers.is_empty() {
                        return Err(MessageBusError::config_error(
                            "kafka.brokers", 
                            "[]", 
                            "Kafka brokers 不能为空"
                        ));
                    }
                    if kafka_config.client_id.is_empty() {
                        return Err(MessageBusError::config_error(
                            "kafka.client_id", 
                            "", 
                            "Kafka 客户端 ID 不能为空"
                        ));
                    }
                } else {
                    return Err(MessageBusError::config_error(
                        "kafka", 
                        "null", 
                        "Kafka 配置缺失"
                    ));
                }
            }
            MessageBusType::Nats => {
                if let Some(ref nats_config) = self.nats {
                    if nats_config.url.is_empty() {
                        return Err(MessageBusError::config_error(
                            "nats.url", 
                            "", 
                            "NATS URL 不能为空"
                        ));
                    }
                } else {
                    return Err(MessageBusError::config_error(
                        "nats", 
                        "null", 
                        "NATS 配置缺失"
                    ));
                }
            }
            MessageBusType::Memory => {
                // 内存模式无需额外验证
            }
        }
        
        Ok(())
    }
    
    /// 合并配置
    pub fn merge(&mut self, other: MessageBusConfig) {
        // 只合并非默认值
        if other.bus_type != MessageBusType::default() {
            self.bus_type = other.bus_type;
        }
        
        if other.max_connections != MessageBusConfig::default().max_connections {
            self.max_connections = other.max_connections;
        }
        
        if other.retry_attempts != MessageBusConfig::default().retry_attempts {
            self.retry_attempts = other.retry_attempts;
        }
        
        if other.enable_metrics != MessageBusConfig::default().enable_metrics {
            self.enable_metrics = other.enable_metrics;
        }
        
        // 合并自定义属性
        for (key, value) in other.custom_properties {
            self.custom_properties.insert(key, value);
        }
        
        // 合并特定配置
        if other.redis.is_some() {
            self.redis = other.redis;
        }
        if other.kafka.is_some() {
            self.kafka = other.kafka;
        }
        if other.nats.is_some() {
            self.nats = other.nats;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::NamedTempFile;
    use std::io::Write;

    #[test]
    fn test_default_config() {
        let config = MessageBusConfig::default();
        assert_eq!(config.bus_type, MessageBusType::Memory);
        assert_eq!(config.max_connections, 100);
        assert!(config.enable_metrics);
        assert!(config.validate().is_ok());
    }

    #[test]
    fn test_config_validation() {
        let mut config = MessageBusConfig::default();
        
        // 测试无效配置
        config.max_connections = 0;
        assert!(config.validate().is_err());
        
        // 修复配置
        config.max_connections = 10;
        assert!(config.validate().is_ok());
    }

    #[test]
    fn test_config_from_env() {
        std::env::set_var("RUSTCLOUD_BUS_TYPE", "redis");
        std::env::set_var("RUSTCLOUD_BUS_MAX_CONNECTIONS", "50");
        std::env::set_var("RUSTCLOUD_BUS_ENABLE_METRICS", "false");
        
        let config = MessageBusConfig::from_env().unwrap();
        assert_eq!(config.bus_type, MessageBusType::Redis);
        assert_eq!(config.max_connections, 50);
        assert!(!config.enable_metrics);
        
        // 清理环境变量
        std::env::remove_var("RUSTCLOUD_BUS_TYPE");
        std::env::remove_var("RUSTCLOUD_BUS_MAX_CONNECTIONS");
        std::env::remove_var("RUSTCLOUD_BUS_ENABLE_METRICS");
    }

    #[test]
    fn test_config_from_yaml_file() {
        let yaml_content = r#"
bus_type: kafka
max_connections: 200
connection_timeout_ms: 5000
retry_attempts: 3
retry_delay_ms: 1000
enable_metrics: true
metrics_port: 9090
enable_tracing: true
buffer_size: 1000
max_message_size: 1048576
message_ttl_ms: 3600000
custom_properties:
  service: "test-service"
  version: "1.0.0"
kafka:
  brokers: ["localhost:9092", "localhost:9093"]
  client_id: "test-client"
  security_protocol: "PLAINTEXT"
  producer:
    acks: "all"
    retries: 3
    batch_size: 16384
    linger_ms: 0
    buffer_memory: 33554432
    compression_type: "none"
  consumer:
    group_id: "test-group"
    auto_offset_reset: "earliest"
    enable_auto_commit: false
    auto_commit_interval_ms: 5000
    session_timeout_ms: 30000
    heartbeat_interval_ms: 3000
    max_poll_records: 500
"#;
        
        // 创建带有正确扩展名的临时文件
        let temp_dir = std::env::temp_dir();
        let temp_path = temp_dir.join("test_config.yaml");
        std::fs::write(&temp_path, yaml_content).unwrap();
        
        let config = MessageBusConfig::from_file(temp_path.to_str().unwrap()).unwrap();
        assert_eq!(config.bus_type, MessageBusType::Kafka);
        assert_eq!(config.max_connections, 200);
        assert!(config.kafka.is_some());
        
        let kafka_config = config.kafka.unwrap();
        assert_eq!(kafka_config.brokers.len(), 2);
        assert_eq!(kafka_config.client_id, "test-client");
        assert_eq!(kafka_config.security_protocol, "PLAINTEXT");
        assert_eq!(kafka_config.producer.acks, "all");
        assert_eq!(kafka_config.producer.linger_ms, 0);
        assert_eq!(kafka_config.consumer.group_id, "test-group");
        assert!(!kafka_config.consumer.enable_auto_commit);
        
        // 验证自定义属性
        assert_eq!(config.custom_properties.get("service"), Some(&"test-service".to_string()));
        assert_eq!(config.custom_properties.get("version"), Some(&"1.0.0".to_string()));
        
        // 清理临时文件
        std::fs::remove_file(&temp_path).ok();
    }

    #[test]
    fn test_config_merge() {
        let mut base_config = MessageBusConfig::default();
        let mut override_config = MessageBusConfig::default();
        
        override_config.max_connections = 500;
        override_config.enable_metrics = false;
        override_config.custom_properties.insert("test".to_string(), "value".to_string());
        
        base_config.merge(override_config);
        
        assert_eq!(base_config.max_connections, 500);
        assert!(!base_config.enable_metrics);
        assert_eq!(base_config.custom_properties.get("test"), Some(&"value".to_string()));
    }
}